package com.mdd.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.mdd.common.BaseContext;
import com.mdd.common.R;
import com.mdd.dto.VoucherOrderDto;
import com.mdd.pojo.User;
import com.mdd.pojo.Voucher;
import com.mdd.pojo.VoucherOrder;
import com.mdd.pojo.request.VoucherOrderAddRequest;
import com.mdd.pojo.request.VoucherOrderPageRequest;
import com.mdd.service.SeckillVoucherService;
import com.mdd.service.UserService;
import com.mdd.service.VoucherOrderService;
import com.mdd.mapper.VoucherOrderMapper;
import com.mdd.service.VoucherService;
import com.mdd.utils.RedisIdWorker;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.mdd.content.RedisContent.GLOBAL_SECKILL_KEY_PREFIX;

/**
* @author 21958
* @description 针对表【voucher_order】的数据库操作Service实现
* @createDate 2023-04-04 23:01:24
*/
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder>
    implements VoucherOrderService{

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private UserService userService;

    @Autowired
    private VoucherService voucherService;

    /**
     * 全局id生成器
     */
    @Autowired
    private RedisIdWorker idWorker;

    /**
     * 生成普通优惠券订单
     * @param voucherOrderAddRequest
     * @return
     */
    @Override
    public R<Long> createVoucherOrder(VoucherOrderAddRequest voucherOrderAddRequest) {
        //检验
        if (voucherOrderAddRequest == null) {
            return R.error("购买失败");
        }
        //解析请求数据
        VoucherOrder voucherOrder = new VoucherOrder();
        BeanUtils.copyProperties(voucherOrderAddRequest,voucherOrder);
        //生成订单
        if (!save(voucherOrder)) {
            return R.error("购买失败");
        }
        //返回订单号
        return R.success(voucherOrder.getId());
    }

    /**
     * 检验是否有购买资格的lua脚本
     */
    private static final DefaultRedisScript<Long> CHECK_SCRIPT;

    /**
     * 检验资格脚本的初始化
     */
    static {
        CHECK_SCRIPT = new DefaultRedisScript<>();
        CHECK_SCRIPT.setLocation(new ClassPathResource("/lua/seckill_check.lua"));
        CHECK_SCRIPT.setResultType(Long.class);
    }

    /**
     * 异步生成订单的线程
     */
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    //@PostConstruct
    private void init(){
        SECKILL_ORDER_EXECUTOR.submit(new SeckillOrderHandle());
    }

    /**
     * 消息队列名
     */
    private String queueName = "stream.orders";
    /**
     * 执行生成订单
     */
    private class SeckillOrderHandle implements Runnable{
        @Override
        public void run() {
            while (true) {
                try {
                    //1.从消息队列中取消息
                    List<MapRecord<String, Object, Object>> read = redisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName,ReadOffset.lastConsumed()));
                    //2.判断是否拿到消息
                    if (read == null || read.isEmpty()) {
                        //继续获取消息
                        continue;
                    }
                    //3.拿到消息则解析消息并处理并生成订单
                    MapRecord<String, Object, Object> msg = read.get(0);
                    Map<Object, Object> value = msg.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    //3.1处理订单
                    voucherOrderHandle(voucherOrder);
                    //4.ack消息确认消息已经被处理
                    redisTemplate.opsForStream().acknowledge(queueName,"g1",msg.getId());
                } catch (Exception e) {
                    log.error("处理订单异常1");
                    handlePendingList();
                }
            }
        }
    }

    /**
     * 处理pendinglist中的消息
     */
    private void handlePendingList() {
        while (true) {
            while (true) {
                try {
                    //1.从消息队列中取消息
                    List<MapRecord<String, Object, Object>> read = redisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName,ReadOffset.from("0")));
                    //2.判断是否拿到消息
                    if (read == null) {
                        // 如果为null，说明没有异常消息，结束循环
                        break;
                    }
                    //3.拿到消息则解析消息并处理并生成订单
                    MapRecord<String, Object, Object> msg = read.get(0);
                    Map<Object, Object> value = msg.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    //3.1处理订单
                    voucherOrderHandle(voucherOrder);
                    //4.ack消息确认消息已经被处理
                    redisTemplate.opsForStream().acknowledge(queueName,"g1",msg.getId());
                } catch (Exception e) {
                    log.error("处理订单异常2");
                }
            }
        }
    }

    /**
     * 生成订单向订单表插入数据
     * @param voucherOrder
     */
    private void voucherOrderHandle(VoucherOrder voucherOrder) {
        //获取锁对象
        RLock lock = redissonClient.getLock("lock:seckill:user:" + voucherOrder.getUserId());
        //尝试获取锁
        boolean b = lock.tryLock();
        if (!b) {
            log.error("不允许重复下单");
            return;
        }
        //创建订单
        createOrder(voucherOrder);
        //释放锁
        lock.unlock();
    }

    @Autowired
    private SeckillVoucherService seckillVoucherService;
    /**
     * 创建订单
     * @param voucherOrder
     */
    @Transactional
    void createOrder(VoucherOrder voucherOrder) {
        Long voucherId = voucherOrder.getVoucherId();
        //扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
                .update();
        if (!success) {
            log.error("库存不足");
        }
        //添加订单
        save(voucherOrder);
    }

    /**
     * 生成秒杀订单
     * @param voucherOrderAddRequest
     * @return
     */
    @Override
    public R<Long> createSeckillOrder(VoucherOrderAddRequest voucherOrderAddRequest) {
        //检验
        if (voucherOrderAddRequest == null) {
            return R.error("数据有误");
        }
        //1.利用redis进行购买权限的校验
        String userId = BaseContext.getCurrentId().toString();
        String voucherId = voucherOrderAddRequest.getVoucherId().toString();
        Long id = idWorker.nextId(GLOBAL_SECKILL_KEY_PREFIX);
        Long res = redisTemplate.execute(CHECK_SCRIPT, Collections.emptyList(), userId, voucherId, id.toString());
        //1.1检验是否有库存
        //1.2检查该用户是否已经抢购过秒杀券
        //1.3若符合抢购条件则返回0，表示具有抢购资格
        if (res != 0) {
            return R.error(res==1?"库存不足!":"请勿重复购买!");
        }
        //2.若等于0则代表可以秒杀
        //返回订单号
        return R.success(id);
    }

    /**
     * 优惠券订单分页
     * @param voucherOrderPageRequest
     * @return
     */
    @Override
    public R<Page<VoucherOrderDto>> voucherOrderPage(VoucherOrderPageRequest voucherOrderPageRequest) {
        //todo 修改
        //检验请求数据
        if (voucherOrderPageRequest == null) {
            return R.error("错误");
        }
        //创建分页对象
        Integer current = voucherOrderPageRequest.getPage();
        Integer pageSize = voucherOrderPageRequest.getPageSize();
        Page<VoucherOrder> voucherOrderPage = new Page<VoucherOrder>(current,pageSize);
        //创建条件构造器
        LambdaQueryWrapper<VoucherOrder> queryWrapper = new LambdaQueryWrapper<>();
        Long id = voucherOrderPageRequest.getId();
        Long userId = voucherOrderPageRequest.getUserId();
        Long voucherId = voucherOrderPageRequest.getVoucherId();
        // todo 先以id userId voucherId 为条件进行查询，后续拓展查询条件。
        queryWrapper.eq(id!=null,VoucherOrder::getId,id)
                .eq(userId!=null,VoucherOrder::getUserId,userId)
                .eq(voucherId!=null,VoucherOrder::getVoucherId,voucherId);
        //分页查询
        Page<VoucherOrder> page = page(voucherOrderPage, queryWrapper);
        //处理数据
        Page<VoucherOrderDto> voucherOrderDtoPage = new Page<>();
        BeanUtils.copyProperties(page,voucherOrderDtoPage,"records");
        List<VoucherOrder> voucherOrderList = page.getRecords();
        ArrayList<VoucherOrderDto> voucherOrderDtos = new ArrayList<>();
        voucherOrderDtoPage.setRecords(voucherOrderDtos);
        for (VoucherOrder voucherOrder : voucherOrderList) {
            VoucherOrderDto voucherOrderDto = new VoucherOrderDto();
            //拷贝
            BeanUtils.copyProperties(voucherOrder,voucherOrderDto);
            //得到用户id
            Long userId1 = voucherOrder.getUserId();
            //得到优惠券id
            Long voucherId1 = voucherOrder.getVoucherId();
            User byId = userService.getById(userId1);
            Voucher byId1 = voucherService.getById(voucherId1);
            //封装到dto中
            voucherOrderDto.setUserName(byId.getName());
            voucherOrderDto.setVoucherName(byId1.getTitle());

            voucherOrderDtos.add(voucherOrderDto);
        }
        //返回数据
        return R.success(voucherOrderDtoPage);
    }
}




